package com.minelx.datamicroscope.common;

import io.gitee.minelx.commontools.factor.CloseableIterator;

import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.function.Consumer;
import java.util.function.Function;

import static java.util.stream.Collectors.toList;
import static java.util.stream.IntStream.range;

// FIXME wait for: replace with Spliterator
public interface EnrichedIterator<E> extends CloseableIterator<E> {
	default void consumeAll(Consumer<E> consumer, int nThreads) {
		ExecutorService executorService = Executors.newFixedThreadPool(nThreads);

		// push workers to executor
		List<? extends Future<?>> tasks = range(0, nThreads)
				.mapToObj(likeThis -> executorService.submit(() -> consumeNextContinuously(consumer)))
				.collect(toList());

		// shutdown executor after execution
		executorService.shutdown();

		// wait til tasks completed
		waitForCompletion(tasks);
	}

	default EnrichedIterator<List<E>> pack(int limit) {
		return new EnrichedIterator<List<E>>() {
			private final EnrichedIterator<E> delegate = EnrichedIterator.this;

			@Override
			public boolean hasNext() {
				return delegate.hasNext();
			}

			@Override
			public List<E> next() {
				List<E> result = new LinkedList<>();
				for (int at = 0; at < limit; at++) {
					if (delegate.hasNext()) {
						result.add(delegate.next());
					}
				}
				return result;
			}

			@Override
			public void close() {
				delegate.close();
			}
		};
	}

	default void consumeNextContinuously(Consumer<E> consumer) {
		while (hasNext()) {
			E result;
			synchronized (this) {
				if (!hasNext()) {
					return;
				}
				result = next();
			}
			consumer.accept(result);
		}
	}

	static void waitForCompletion(List<? extends Future<?>> tasks) {
		tasks.forEach(future -> {
			try {
				future.get();
			} catch (Exception e) {
				// cancel all tasks
				tasks.forEach(task -> task.cancel(true));
				throw new RuntimeException("error while executing tasks.", e);
			}
		});
	}

	default <R> EnrichedIterator<R> map(Function<E, R> mapper) {
		return new EnrichedIterator<R>() {
			@Override
			public R next() {
				return mapper.apply(EnrichedIterator.this.next());
			}

			@Override
			public boolean hasNext() {
				return EnrichedIterator.this.hasNext();
			}

			@Override
			public void close() {
				EnrichedIterator.this.close();
			}
		};
	}

	static <T> EnrichedIterator<T> fromUnSafeIterator(Iterator<T> iterator) {
		return fromIterator(new SafeIterator<T>(iterator));
	}

	static <T> EnrichedIterator<T> fromIterator(Iterator<T> iterator) {
		return new EnrichedIterator<T>() {
			@Override
			public T next() {
				return iterator.next();
			}

			@Override
			public boolean hasNext() {
				return iterator.hasNext();
			}

			@Override
			public void close() {
				// do nothing
			}
		};
	}

	static <T> EnrichedIterator<T> fromCloseableIterator(CloseableIterator<T> iterator) {
		return new EnrichedIterator<T>() {
			@Override
			public T next() {
				return iterator.next();
			}

			@Override
			public boolean hasNext() {
				return iterator.hasNext();
			}

			@Override
			public void close() {
				iterator.close();
			}
		};
	}

	class SafeIterator<E> implements Iterator<E> {
		private final Iterator<E> delegator;

		public SafeIterator(Iterator<E> delegator) {
			this.delegator = delegator;
		}

		@Override
		public synchronized boolean hasNext() {
			return delegator.hasNext();
		}

		@Override
		public synchronized E next() {
			return delegator.next();
		}
	}
}